package com.qqt.csr.archive.batch.config;

import com.qqt.csr.archive.batch.task.context.ConversationArchiveTaskContext;
import com.qqt.csr.archive.batch.task.context.MessageArchiveTaskContext;
import com.qqt.csr.archive.batch.task.etl.reader.ConversationArchiveReader;
import com.qqt.csr.archive.batch.task.etl.reader.MessageArchiveReader;
import com.qqt.csr.archive.batch.task.etl.tasklet.AccountPreTasklet;
import com.qqt.csr.archive.batch.task.etl.transformer.ConversationArchiveProcessor;
import com.qqt.csr.archive.batch.task.etl.transformer.MessageArchiveProcessor;
import com.qqt.csr.archive.batch.task.etl.writer.ConversationArchiveWriter;
import com.qqt.csr.archive.batch.task.etl.writer.MessageArchiveWriter;
import com.qqt.csr.archive.batch.task.exception.ArchiveTaskExceptionHandler;
import com.qqt.csr.archive.entity.writer.ArchiveConversation;
import com.qqt.csr.archive.entity.writer.ArchiveMessage;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.retry.RetryPolicy;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;

import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author: yang hao
 * @Date: 2024/11/23 9:15
 * @Description: 归档任务配置
 **/
@Configuration
@EnableBatchProcessing(isolationLevelForCreate = "ISOLATION_REPEATABLE_READ")
@SuppressWarnings("unused")
public class ArchiveTaskConfig {

    /**
     * 线程池的参数
     */
    @Value("${task.coreSize:4}")
    private Integer coreSize;
    @Value("${task.maxSize:4}")
    private Integer maxSize;
    @Value("${task.queueSize:200}")
    private Integer queueSize;
    @Value("${task.batchSize:200}")
    private Integer batchSize;
    /**
     * 任务信息 仓库
     */
    @Autowired
    private JobRepository jobRepository;
    /**
     * 事务管理器
     */
    @Autowired
    private PlatformTransactionManager transactionManager;
    /**
     * 异常处理器
     */
    @Autowired
    private ArchiveTaskExceptionHandler archiveTaskExceptionHandler;
    /**
     * 预处理任务
     */
    @Autowired
    private AccountPreTasklet accountPreTasklet;
    /**
     * 归档读取
     */
    @Autowired
    private ConversationArchiveReader conversationArchiveReader;
    @Autowired
    private MessageArchiveReader messageArchiveReader;
    /**
     * 归档处理
     */
    @Autowired
    private ConversationArchiveProcessor conversationArchiveProcessor;
    @Autowired
    private MessageArchiveProcessor messageArchiveProcessor;
    /**
     * 归档写入
     */
    @Autowired
    private ConversationArchiveWriter conversationArchiveWriter;
    @Autowired
    private MessageArchiveWriter messageArchiveWriter;

    /**
     * 配置线程池
     * @return {@link ThreadPoolTaskExecutor} 线程池
     */
    @Bean
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(coreSize);
        executor.setMaxPoolSize(maxSize);
        executor.setQueueCapacity(queueSize);
        executor.setThreadNamePrefix("batch-thread-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }

    @Bean
    public RetryPolicy retryPolicy() {
        SimpleRetryPolicy policy = new SimpleRetryPolicy();
        // 最大重试次数
        policy.setMaxAttempts(3);
        return policy;
    }

    @Bean
    public ExponentialBackOffPolicy backOffPolicy() {
        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        // 设置初始退避时间间隔为1秒 ex: 1s, 3s, 9s
        backOffPolicy.setInitialInterval(1000);
        // 设置退避倍数
        backOffPolicy.setMultiplier(3.0);
        // 设置最大退避时间间隔为10秒
        backOffPolicy.setMaxInterval(10000);
        return backOffPolicy;
    }


    /**
     * 配置消息归档步骤
     * @return {@link Step} 消息归档步骤
     */
    @Bean
    public Step messageArchiveStep() {
        return new StepBuilder("messageArchiveTask", jobRepository)
                .<MessageArchiveTaskContext, List<ArchiveMessage>>chunk(batchSize, transactionManager)
                .reader(messageArchiveReader)
                .processor(messageArchiveProcessor)
                .writer(messageArchiveWriter)
                //.taskExecutor(taskExecutor())
                .faultTolerant()
                .retryPolicy(retryPolicy())
                .backOffPolicy(backOffPolicy())
                .exceptionHandler(archiveTaskExceptionHandler)
                .build();
    }

    /**
     * 配置会话归档步骤
     * @return {@link Step} 会话归档步骤
     */
    @Bean
    public Step conversationArchiveStep() {
        return new StepBuilder("conversationArchiveStep", jobRepository)
                .<ConversationArchiveTaskContext, ArchiveConversation>chunk(batchSize, transactionManager)
                .reader(conversationArchiveReader)
                .processor(conversationArchiveProcessor)
                .writer(conversationArchiveWriter)
                //.taskExecutor(taskExecutor())
                .faultTolerant()
                .retry(Exception.class)
                .retryPolicy(retryPolicy())
                .backOffPolicy(backOffPolicy())
                .exceptionHandler(archiveTaskExceptionHandler)
                .build();
    }

    /**
     * 配置预处理步骤
     * @return {@link Step} 预处理步骤
     */
    @Bean
    public Step accountPreStep() {
        return new StepBuilder("accountPreStep", jobRepository)
                .tasklet(accountPreTasklet, transactionManager)
                //.taskExecutor(taskExecutor())
                .exceptionHandler(archiveTaskExceptionHandler)
                .build();
    }

    /**
     * 配置归档任务
     * @return {@link Job} 归档任务
     */
    @Bean
    public Job archiveJob() {
        return new JobBuilder("archiveJob", jobRepository)
                .start(accountPreStep())
                .next(conversationArchiveStep())
                .next(messageArchiveStep())
                .build();
    }

}
